#include "udp_source.h"
#include "ts_stream_demuxer.h"

udpSource::udpSource()
{
	m_listen_thread = NULL;
	m_thread_pool = NULL;
	m_listen_thread_stop = true;
	m_cur_connect_num = 0;
	m_max_connect_num = m_thread_pool_size = DEFULT_MAX_CONNECT_NUM;
	m_port = DEFULT_UDP_SERVER_PORT;
	m_client_timeout = 30000;//milliseconds

}

udpSource::~udpSource()
{

}

int udpSource::udpServerInit()
{
	GError *err = NULL;
	m_thread_pool = g_thread_pool_new (threadPoolFunc, this, m_thread_pool_size, TRUE, &err);
	if (err != NULL)
	{
		printf ("Create thread pool error %s", err->message);
	    g_error_free (err);
	    return -1;
	}

	struct sockaddr_in serverAddr;
	 /* 开启 socket 监听 */
	if ((m_listener = socket(PF_INET, SOCK_DGRAM, 0)) == -1)
	{
		perror("socket create failed ！");
		return -1;
	}
	else
	{
		printf("socket create  success \n");
	}

	if(m_client_list == NULL)
	{
		m_client_list = new udpClientChannel[DEFULT_MAX_CONNECT_NUM];
	}


	/*设置socket属性，端口可以重用*/
	int opt=SO_REUSEADDR;
	setsockopt(m_listener,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
	if (fcntl(m_listener, F_SETFL, fcntl(m_listener, F_GETFD, 0)|O_NONBLOCK) == -1)
	{
		return -1;
	}

	bzero(&serverAddr, sizeof(serverAddr));
	serverAddr.sin_family = PF_INET;
	serverAddr.sin_port = htons(m_port);
	serverAddr.sin_addr.s_addr = INADDR_ANY;
	if (bind(m_listener, (struct sockaddr *) &serverAddr, sizeof(struct sockaddr)) == -1)
	{
		perror("bind");
		return -1;
	}
	else
	{
		printf("IP and port bind success \n");
	}

	/* 创建 epoll 句柄，把监听 socket 加入到 epoll 集合里 */
	m_epfd = epoll_create(MAX_EPOLL_SIZE);
	m_ev.events = EPOLLIN;// | EPOLLET;
	m_ev.data.fd = m_listener;
	if (epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_listener, &m_ev) < 0)
	{
		fprintf(stderr, "epoll set insertion error: fd=%d\n", m_listener);
		return -1;
	}
	else
	{
		printf("listen socket added in  epoll success \n");
	}
	return 0;
}

int udpSource::udpServerStart()
{
	if(udpServerInit() == 0)
	{
		m_listen_thread_stop = false;
		m_listen_thread = g_thread_new ("listen_thread", listenThreadFunc, this);
		return 0;
	}
	return -1;
}
int udpSource::udpServerStop()
{
	if(m_listen_thread)
	{
		m_listen_thread_stop = true;
		g_thread_join (m_listen_thread);
		close(m_epfd);
		close(m_listener);
		g_thread_pool_free(m_thread_pool,false,false);
		m_listen_thread = NULL;
		m_thread_pool = NULL;
	}
	return 0;
}
void udpSource::threadPoolFunc(void* data,void*user_data)
{
	udpSource *pSObj = (udpSource*)user_data;
	udpClientChannel *pCObj = (udpClientChannel*)data;

	int clientTimeout = 0;
	uint8_t * es_out = NULL;
	int es_len = 0;
	uint32_t es_timestamp = 0;
	stream_type::stype type;
	tsde_status::status tsde_res;
	tsDemuxer ts_demuxer;

#ifdef WRITE_TS_FILE
	char *pStreamName = pCObj->udpClientGetStreamName();
	char fileName[256];
	sprintf(fileName,"./%s.ts",pStreamName+21);
	pCObj->m_ts_file = fopen(fileName,"wb");
#endif

	printf("%s\n",fileName);
	while(pCObj->udpClientIsRun())
	{
		ts_dd * tsData = pCObj->udpClientGetTsData();
		if(tsData == NULL)
		{
			usleep(1000);
			clientTimeout++;
			if(clientTimeout >= pSObj->m_client_timeout)
			{
				pCObj->udpClientSetState(false);
				printf("%s--->timeout\n",pStreamName);
				break;
			}
			else
			{
				continue;
			}
		}
		clientTimeout = 0;

		int pkg_num = (tsData->size - 5) / 188;
		const char * ptr = (char*)tsData->data + 5;
		while(pkg_num--)
		{
			tsde_res = ts_demuxer.demuxATsPacket(ptr,&es_out,&es_len,&es_timestamp,&type);

			if(tsde_res == tsde_status::status_demux_ok && type == stream_type::aac_audio)
			{
				//printf("%d\n",tsde_res);
				fwrite(es_out,1,es_len,pCObj->m_ts_file);
			}
			ptr += 188;
		}

#ifdef WRITE_TS_FILE
		//fwrite(tsData->data+5,tsData->size-5,1,pCObj->m_ts_file);
#endif
		delete [] tsData->data;
		delete tsData;

	}

	pCObj->udpClientFflushQueue();

#ifdef WRITE_TS_FILE
	if(pCObj->m_ts_file)
	{
		fclose(pCObj->m_ts_file);
		pCObj->m_ts_file = NULL;
	}
#endif
}
void* udpSource::listenThreadFunc (void* data)
{
	udpSource *pObj = (udpSource*)data;
	int nfds;
	int n;
	int recvLen;
	GError *err = NULL;
	uint8_t recvBuf[MAX_REC_BUFFER + 1];
	struct sockaddr_in client_addr;
	socklen_t cli_len=sizeof(client_addr);

	while (!pObj->m_listen_thread_stop)
	{
	    /* 等待有事件发生 */
	    nfds = epoll_wait(pObj->m_epfd, pObj->m_events, pObj->m_max_connect_num, 300);
	    if (nfds == -1)
	    {
	    	perror("epoll_wait");
	    	break;
	    }

	    /* 处理所有事件 */
	    for (n = 0; n < nfds; ++n)
	    {
	    	if (pObj->m_events[n].data.fd == pObj->m_listener)
	    	{

	    		/* 开始处理每个新连接上的数据收发 */
	    		//bzero(recvBuf, MAX_REC_BUFFER + 1);

	    		/* 接收客户端的消息 */
	    		//printf("%d\n",n);
	    		recvLen = recvfrom(pObj->m_listener, recvBuf, MAX_REC_BUFFER, 0, (struct sockaddr *)&client_addr, &cli_len);
	    		if (recvLen > 0)
	    		{
	    			int key = pObj->udpServerHash(&client_addr) % pObj->m_max_connect_num;
	    			char pkgType = udpTsProtocolParser::getPkgType(recvBuf);
	    			//printf("hash:%d   %c\n",key,pkgType);
	    			if(pkgType == START_PUSH_STREAM)
	    			{printf("hash:%d\n",key);
	    				if(pObj->m_client_list[key].udpClientIsRun())
	    				{
	    					printf("hash error\n");
	    				}
	    				char* pStreamName = pObj->m_client_list[key].udpClientGetStreamName();
	    				int ind = sprintf(pStreamName,"%s:%d#",inet_ntoa(client_addr.sin_addr),ntohs(client_addr.sin_port));

	    				if(recvLen - 1 > MAX_STREAM_NAME_LEN)
	    				{
	    					recvLen = MAX_STREAM_NAME_LEN + 1;
	    				}

	    				udpTsProtocolParser::getStreamName(recvBuf,recvLen,pStreamName+ind);

	    				pObj->m_client_list[key].udpClientSetState(true);
	    				pObj->m_cur_connect_num++;
	    				g_thread_pool_push (pObj->m_thread_pool, &pObj->m_client_list[key], &err);
	    				if (err != NULL)
	    				{
	    					printf ("Thread pool push error %s", err->message);
	    					g_error_free (err);
	    				}
	    				printf("connect_num:%d     %s\n",pObj->m_cur_connect_num,pStreamName);
	    			}
	    			else if(pkgType == PUSH_STREAM_DATA)
	    			{
	    				if(pObj->m_client_list[key].udpClientIsRun())
	    				{
	    					uint8_t * ts_data = new uint8_t[recvLen];
	    					ts_dd * ts = new ts_dd();
	    					ts->data = ts_data;
	    					ts->size = recvLen;
	    					memcpy(ts_data,recvBuf,recvLen);
	    					pObj->m_client_list[key].udpClientPushTsData(ts);
	    				}
	    			}
	    			else if(pkgType == STOP_PUSH_STREAM)
	    			{
	    				if(pObj->m_client_list[key].udpClientIsRun())
	    				{
	    					pObj->m_client_list[key].udpClientSetState(false);
	    				}
	    			}

	    		}
	    		else
	    		{
	    			printf("error \n");
	    		}
	    	}
	    }
	}
	return NULL;
}
uint32_t udpSource::udpServerHash(struct sockaddr_in *addr)
{
	char buff[20],*str = buff;
	sprintf(buff,"%s:%d#",inet_ntoa(addr->sin_addr),ntohs(addr->sin_port));

	uint32_t seed = 131; //31,131,1313,131313 ...
	uint32_t hashd = 0;
	while(*str)
	{
		hashd = hashd * seed + (*str++);
	}

	return (hashd & 0x7FFFFFFF);

}
int udpSource::udpServerConfig(const char* config_file)
{
	return 0;
}
